package dataStream;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import dataStream.Person;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc 滑动窗口: 每间隔 slice 的时间，统计前 size 时间的数据，如每间隔 1min 统计前 5 分钟的数据
 * @Author xuyouchang
 * @Date 2021/10/9
 */
public class SlidingWindow {
    public static void main(String[] args) {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream = env.fromCollection(data);
        // 事件时间 window
        dataStream.keyBy(person->person.gender).window(SlidingEventTimeWindows.of(Time.seconds(1000L), Time.seconds(10L)));

        // 处理时间 window
        dataStream.keyBy(person->person.gender).window(SlidingProcessingTimeWindows.of(Time.seconds(1000L), Time.seconds(10L)));


    }
}
